package spark.master

import akka.actor.{Actor, ActorRef, ActorSystem, Props}
import com.typesafe.config.ConfigFactory
import spark.common.{HeartBeat, RegisterWorkerInfo, RegisteredWorkerInfo, RemoveTimeOutWorker, StartTimeOutWorker, WorkerInfo}

import scala.collection.mutable
import scala.concurrent.duration._

/**
 * Created by DJJ on 2019/12/2.
 */
class SparkMaster extends Actor {
  var map=mutable.Map[String,WorkerInfo]()
  override def receive: Receive = {
    case "start" => {
      println("master 启动")
      self ! StartTimeOutWorker
    }

    case  RegisterWorkerInfo(id,cpu,ram)=>{
      if (!map.contains(id)){
        map +=((id,new WorkerInfo(id,cpu,ram)))
      }
      sender() ! RegisteredWorkerInfo
    }
    case HeartBeat(id)=>{
      println("worker = " + id + "更新心跳")
      val workerInfo = map(id)
      workerInfo.lastHeartBeat=System.currentTimeMillis()
    }
    case StartTimeOutWorker=>{
      import context.dispatcher
      /**
       * 1, 0 millis 发送不延时
       * 2, 9000 millis 每隔9秒执行一次
       * 3,self 发送给自己
       * 4, RemoveTimeOutWorker 发送的内容
       */
      context.system.scheduler.schedule(0 millis, 9000 millis, self, RemoveTimeOutWorker)
    }
    case RemoveTimeOutWorker=>{
      val workerInfos = map.values
      //now - heartBeat>6000
      workerInfos.filter(w=>(System.currentTimeMillis()-w.lastHeartBeat)>6000).foreach(info=>{
        map.remove(info.id)
      })
      println("存货worker数量:"+map.size)
    }
  }
}

object SparkMaster extends App {
  val host = "127.0.0.1" //服务端ip地址
  val port = 10005
  //创建config对象,指定协议类型，监听的ip和端口
  val config = ConfigFactory.parseString(
    s"""
       |akka.actor.provider="akka.remote.RemoteActorRefProvider"
       |akka.remote.netty.tcp.hostname=$host
       |akka.remote.netty.tcp.port=$port
        """.stripMargin)
  val sparkMaster = ActorSystem("SparkMaster", config)
  val master: ActorRef = sparkMaster.actorOf(Props[SparkMaster], "SparkMaster-01")
  master ! "start"

}
